package com.cloud.mqtt.Emq;

/**
 * @program: mqtt
 * @description:
 * 发布消息的回调类
 *
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
 * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
 * 在回调中，将它用来标识已经启动了该回调的哪个实例。
 * 必须在回调类中实现三个方法：
 *
 *  public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
 *
 *  public void connectionLost(Throwable cause)在断开连接时调用。
 *
 *  public void deliveryComplete(MqttDeliveryToken token))
 *  接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
 *  由 MqttClient.connect 激活此回调。
 * @author: Cloud
 * @create: 2020/11/21 10:43:51
 */
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cloud.iotdb.utils.IotDBDao;
import com.cloud.mqtt.util.DateUtils;
import com.cloud.mqtt.util.SpringContextUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.*;

@Slf4j
public class PushCallback implements MqttCallback {

    private MQTTSubscribe mqttSubscribe;

    private MqttConfig mqttConfig;

    public PushCallback(MQTTSubscribe subscribe) {
        this.mqttSubscribe = subscribe;
    }

    public void connectionLost(Throwable throwable) {
        // 连接丢失后，一般在这里面进行重连
        log.info("---------------------连接断开，正在重连...---------------------");

        while (true) {
            // 客户端重连
            try {
                throwable.printStackTrace();
                //如果没有发生异常说明连接成功，如果发生异常，则死循环
                Thread.sleep(1000);
                MQTTServer mqttServer = SpringContextUtils.getBean(MQTTServer.class);
                //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
                mqttServer.connect();
            } catch (Exception e) {
                continue;
            }
        }
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
    }


    //  如何 将messageArrived 中的 result 引用到MQTTServer中还是一个问题
//    public String getMessage_Subsrib_To_Publish(String message) {
//        System.out.println("Message_Subsrib_To_Publish11111:   " + message);
//        return message;
//    }


    /*
     * 功能描述:这里把从网关采集上来的设备的数据拆分并加入到iotdb中
     * @param topic
     * @param message
     * @return void
     * @author cloud
     * @date  2021/10/27 15:50
     */
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面
        String result = new String(message.getPayload(),"UTF-8");
        Date date = new Date();
        String dateFormat = DateUtils.format(date, DateUtils.DATE_TIME_PATTERN[0]);
        log.info("接收消息主题 : " + topic);
        log.info("接收消息Qos : " + message.getQos());
        log.info("接收消息内容 : " + result);
        JSONObject jsonObject = JSON.parseObject(result);
        Object object = jsonObject.get("metrics");
        JSONObject metrics = JSON.parseObject(object.toString());
        List<String> deviceList = new ArrayList<>();
        deviceList.add("A");
        deviceList.add("B");
        deviceList.add("C");
        deviceList.add("D");
        deviceList.add("E");
        List<Map<String, String>> pointList = new ArrayList<>();
        Map<String,String> point = new HashMap<>();
        point.put("status",metrics.get("A").toString());
        point.put("GasStateTime", dateFormat);
        pointList.add(point);
        point = new HashMap<>();
        point.put("status",metrics.get("B").toString());
        point.put("GasStateTime", dateFormat);
        pointList.add(point);
        point = new HashMap<>();
        point.put("status",metrics.get("C").toString());
        point.put("GasStateTime", dateFormat);
        pointList.add(point);
        point  = new HashMap<>();
        point.put("status",metrics.get("D").toString());
        point.put("GasStateTime", dateFormat);
        pointList.add(point);
        point = new HashMap<>();
        point.put("status",metrics.get("E").toString());
        point.put("GasStateTime", dateFormat);
        pointList.add(point);
        IotDBDao iotDBDao = SpringContextUtils.getBean(IotDBDao.class);
        iotDBDao.batchInset("root.device.getAction",deviceList,date.getTime(),pointList);
        log.info(result+"==>>加入到数据成功!");

        MQTTServer mqttServer = SpringContextUtils.getBean(MQTTServer.class);

        //这里可以针对收到的消息做处理
//        mqttServer.sendMessage(mqttConfig.pubTopic,result);
    }

}

